package com.gvtv.main.execute;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import com.gvtv.main.email.EmailInfo;
import com.gvtv.main.email.EmailThread;
import com.gvtv.main.execute.ExecuteJob.ExecuteJobState;
import com.gvtv.main.schedul.SchedulEnv;
import com.gvtv.main.schedul.SchedulInfo;
import com.gvtv.main.util.PropertiesHelper;

public class ExecuteJobManager {
	private static final Logger logger = Logger.getLogger(ExecuteJobManager.class);
	
	private List<ExecuteJob> allJobs = new LinkedList<ExecuteJob>();
	private List<ExecuteJob> successfulJobs = new LinkedList<ExecuteJob>();
	private List<ExecuteJob> failedJobs = new LinkedList<ExecuteJob>();
	
	private ExecutorService executorService;
	private List<Future<ExecuteJob>> futures = new ArrayList<Future<ExecuteJob>>();
	
	private boolean isRunning = true;
	private SchedulInfo schedulInfo;
	private SchedulEnv schedulEnv;
	
	public ExecuteJobManager(SchedulInfo schedulInfo, Collection<ExecuteJob> executeJobs){
		this.schedulInfo = schedulInfo;
		this.schedulEnv = new SchedulEnv(this.schedulInfo.getId());
		this.schedulEnv.add(this.schedulInfo.getEnv());
		addJobs(executeJobs);
	}
	public void addJob(ExecuteJob executeJob){
		executeJob.setJobState(ExecuteJobState.WAITING);
		allJobs.add(executeJob);
	}
	
	public void addJobs(Collection<ExecuteJob> executeJobs){
		for (ExecuteJob executeJob : executeJobs) {
			addJob(executeJob);
		}
	}
	
	private void submitJob(ExecuteJob executeJob){
		executeJob.setJobState(ExecuteJobState.READY_RUNNING);
		executeJob.setSubmitTime(new Date());
		JobWorker jobWorker = null;
		String jobClass = JobWorkerMapping.getInstance().getMapping(executeJob.getJobType());
		try {
			
			jobWorker = (JobWorker) Class.forName(jobClass).getDeclaredConstructor(SchedulInfo.class, ExecuteJob.class, SchedulEnv.class)
					.newInstance(schedulInfo , executeJob, schedulEnv);
			futures.add(executorService.submit(jobWorker));
			logger.info("提交调度["+schedulInfo.getId()+"]的job["+executeJob.getId()+"]到线程池，等待运行。");
		} catch (IllegalArgumentException e) {
			logger.error(e);
		} catch (SecurityException e) {
			logger.error(e);
		} catch (InstantiationException e) {
			logger.error(e);
		} catch (IllegalAccessException e) {
			logger.error(e);
		} catch (InvocationTargetException e) {
			logger.error(e);
		} catch (NoSuchMethodException e) {
			logger.error(e);
		} catch (ClassNotFoundException e) {
			logger.error(jobClass+"类没有找到. "+e);
		}
		if(jobWorker==null && StringUtils.isBlank(jobClass)){
			executeJob.setJobState(ExecuteJobState.FAILED);
			logger.info("构造JobWorker失败，["+executeJob.getId()+"]运行失败。");
		}
	}
	
	public void execute(){
		executorService = Executors.newFixedThreadPool(PropertiesHelper.getInteger("schedul.thread.pool.size", 3));
		logger.info(schedulInfo.getId()+"调度开始运行");
		while (true) {
			Iterator<ExecuteJob> it = allJobs.iterator();
			while(it.hasNext()) {
				ExecuteJob job = it.next();
				switch (job.checkJobState()) {
				case READY:
					submitJob(job);
					break;
				case SUCCESS:
					successfulJobs.add(job);
					it.remove();
					break;
				case FAILED:
				case DEPENDENT_FAILED:
					failedJobs.add(job);
					logger.info("job["+job.getId()+"]运行失败.");
					it.remove();
					break;
				case WAITING:
				case RUNNING:
				case READY_RUNNING:
					break;
				default:
					break;
				}
			}
			if(allJobs.isEmpty()){
				boolean isAllDone = true;
				for(Future<ExecuteJob> future: futures){
					if(!future.isDone()){
						isAllDone = false;
						break;
					}
				}
				if(isAllDone){
					stop();
				}
			}
			if(!isRunning){
				break;
			}
			
			try {
				Thread.sleep(5*1000);
			} catch (InterruptedException e) {
				
			}
		}
	}
	
	public void stop(){
		isRunning = false;
		if(executorService!=null){
			executorService.shutdown();
		}
		sendEmail();
		schedulEnv.clear();
		schedulEnv = null;
		logger.info(schedulInfo.getId()+"调度结束运行");
	}
	
	private void sendEmail(){
		List<ExecuteJob> jobs = new LinkedList<ExecuteJob>();
		if(!successfulJobs.isEmpty()){
			jobs.addAll(successfulJobs);
		}
		if(!failedJobs.isEmpty()){
			jobs.addAll(failedJobs);
		}
		if(!jobs.isEmpty()){
			if(PropertiesHelper.getBoolean("mail.enable", false)){
				EmailInfo emailInfo = new EmailInfo(schedulInfo.getId(), jobs);
				EmailThread.getInstance().addEmailInfo(emailInfo);
			}
		}
	}

}
